package com.angus.modules.content.controller.task;

import com.alibaba.nacos.client.naming.utils.CollectionUtils;
import com.angus.api.front.factory.RemoteFrontUserFallbackFactory;
import com.angus.common.core.enums.ContentType;
import com.angus.common.core.utils.StringUtils;
import com.angus.modules.content.domain.Course;
import com.angus.modules.content.service.ICourseService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

import javax.annotation.PostConstruct;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;

@RestController
@RequestMapping("/task")
public class TaskController {

    private static final Logger log = LoggerFactory.getLogger(TaskController.class);

    // 使用 Sinks 处理动态任务流（线程安全）
    private final Sinks.Many<Runnable> taskSink = Sinks.many().unicast().onBackpressureBuffer();

    // 定义弹性调度器（核心线程数自适应，最大2个线程）
    // 最大线程数
    //  - threadCap 2
    // 队列容量设为合理值（如100），避免OOM
    //  - queuedTaskCap 10
    private final Scheduler boundedScheduler = Schedulers.newBoundedElastic(
            2,
            10,
            "task-pool"
    );

    @Autowired
    private ICourseService courseService;

    @PostConstruct
    public void init() {
        taskSink.asFlux()
            // 允许同时处理最多2个任务
            .flatMap(task -> Mono.fromRunnable(task).subscribeOn(boundedScheduler), 2)
            .subscribe(unused -> {}, ex -> log.error("任务处理失败", ex));
    }

    // 触发异步任务（示例接口）
    @GetMapping("/async-cn-gen-question")
    public Mono<String> triggerCnAsyncTasks(@RequestParam(required = false, defaultValue = "") String courseCode) {
        submitTask(() -> cnRun(courseCode));
        return Mono.just("任务已提交");
    }

    // 触发异步任务（示例接口）
    @GetMapping("/async-en-gen-question")
    public Mono<String> triggerEnAsyncTasks(@RequestParam(required = false, defaultValue = "") String courseCode) {
        submitTask(() -> enRun(courseCode));
        return Mono.just("任务已提交");
    }

    // 提交任务到队列
    private void submitTask(Runnable task) {
        taskSink.emitNext(task, (signalType, emitResult) -> {
            log.error("任务提交失败，队列已满");
            return emitResult == Sinks.EmitResult.FAIL_NON_SERIALIZED;
        });
    }

    // 获取课程编码列表
    private List<String> fetchAllCourseCodes(String type) {
        List<Course> courseList = courseService.getCourseByType(type);
        return courseList.stream().map(Course::getCourseCode).filter(Objects::nonNull).collect(Collectors.toList());
    }

    // 中文任务处理逻辑
    private void cnRun(String courseCode) {
        log.info("cn 处理任务开始，线程: {}", Thread.currentThread().getName());
        List<String> courseCodes = fetchAllCourseCodes(ContentType.CN.getTypeCode());
        if (CollectionUtils.isEmpty(courseCodes)) {
            log.info("未找到中文课程代码，任务结束");
            return;
        }
        List<String> targetCodes;
        if (StringUtils.isNotBlank(courseCode)) {
            targetCodes = courseCodes.stream()
                    .filter(code -> code.contains(courseCode))
                    .collect(Collectors.toList());
            log.info("筛选后中文课程数量: {}", targetCodes.size());
        } else {
            targetCodes = courseCodes;
            log.info("全量中文课程数量: {}", targetCodes.size());
        }
        // 分批处理（示例：每100个课程一批）
        int batchSize = 100;
        for (int i = 0; i < targetCodes.size(); i += batchSize) {
            List<String> batch = targetCodes.subList(i, Math.min(i + batchSize, targetCodes.size()));
            batch.forEach(code ->
                    courseService.GenCourseExercises(code, ContentType.CN.getTypeCode())
            );
            log.info("中文课程已处理批次 {}/{}", i / batchSize + 1, (targetCodes.size() + batchSize - 1) / batchSize);
        }
        log.info("中文课程处理任务完成");
    }

    // 英文任务处理逻辑
    private void enRun(String courseCode) {
        log.info("en 处理任务，线程: {}", Thread.currentThread().getName());
        List<String> courseCodes = fetchAllCourseCodes(ContentType.EN.getTypeCode());
        if (CollectionUtils.isEmpty(courseCodes)) {
            log.info("未找到英文课程代码，任务结束");
            return;
        }
        List<String> targetCodes;
        if (StringUtils.isNotBlank(courseCode)) {
            targetCodes = courseCodes.stream()
                    .filter(code -> code.contains(courseCode))
                    .collect(Collectors.toList());
            log.info("筛选后英文课程数量: {}", targetCodes.size());
        } else {
            targetCodes = courseCodes;
            log.info("全量英文课程数量: {}", targetCodes.size());
        }
        // 分批处理（示例：每100个课程一批）
        int batchSize = 100;
        for (int i = 0; i < targetCodes.size(); i += batchSize) {
            List<String> batch = targetCodes.subList(i, Math.min(i + batchSize, targetCodes.size()));
            batch.forEach(code ->
                    courseService.GenCourseExercises(code, ContentType.EN.getTypeCode())
            );
            log.info("英文课程已处理批次 {}/{}", i / batchSize + 1, (targetCodes.size() + batchSize - 1) / batchSize);
        }
        log.info("英文课程处理任务完成");
    }
}